use msgbus::{bus};
use msgbus::handler::MessageHandler;
use msgbus::message::{Message, CallReturn};
use msgbus::capi::get_bus;
use std::os::raw::c_char;
use std::io::Write;

extern "C" {
    pub fn start_subscribe(topic: *const c_char);
    pub fn stop_subscribe();
}

struct Client;

impl MessageHandler for Client {
    fn handle(&self, msg: &Message) {
        println!("rust: topic {}, content {}", msg.topic(), String::from_utf8_lossy(msg.content()).to_string());
        let _ = std::io::stdout().flush();
    }

    fn call(&self, topic: &str, method: &str, _args: &[u8]) -> CallReturn {
        println!("rust: topic {} call {}", topic, method);
        Ok(Vec::new())
    }
}

fn main() {
    let client = Box::new(Client{});
    let token = bus().subscribe("aa/bb", Some(client), None);
    unsafe { start_subscribe(b"aa/bb\0".as_ptr() as *const c_char); }
    let _ = get_bus();
    std::thread::spawn(||{
        for i in 0..2000 {
            bus().publish("aa/bb",format!("hellow world {}", i).as_bytes(), 1000);
            // if i % 1000 == 0 {
            //     std::thread::sleep(std::time::Duration::from_millis(1));
            // }
        }
        for i in 0..2000 {
            let res = bus().call("aa/bb", "echo",format!("hellow world {}", i).as_bytes(), 1000);
            if let Ok(data) = res {
                let s = String::from_utf8_lossy(data.as_slice());
                println!("call return: {}", s);
            }
            // if i % 1000 == 0 {
            //     std::thread::sleep(std::time::Duration::from_millis(1));
            // }
        }
    });
    std::thread::sleep(std::time::Duration::from_millis(500));
    bus().unsubscribe(token.as_str());
    loop {
        std::thread::sleep(std::time::Duration::from_secs(1));
    }
    // unsafe { stop_subscribe(); }
}
